package thss.mro.enterance;

import thss.mro.bolt.DataStore;
import thss.mro.bolt.PacketBodyParse;
import thss.mro.bolt.PacketHeadParse;
import thss.mro.bolt.PacketIdentify;
import thss.mro.spout.SCPSpout;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;

public class TestMain {

	public final static String ID_SPOUT_SCP = "SCPSpout"; 
	public final static String ID_SPOUT_CELL = "CellSpout"; 
	
	/**
	 * @param args
	 * @throws InvalidTopologyException 
	 * @throws AlreadyAliveException 
	 */
	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout(ID_SPOUT_SCP, new SCPSpout(), 1);    
		builder.setBolt("PacketIdentify", new PacketIdentify(), 1)
		        .shuffleGrouping(ID_SPOUT_SCP)
				.shuffleGrouping("PacketBodyParse","cellStream");
		builder.setBolt("PacketHeadParse", new PacketHeadParse(), 1)
				.shuffleGrouping("PacketIdentify");
		builder.setBolt("PacketBodyParse", new PacketBodyParse(), 1)
				.shuffleGrouping("PacketHeadParse");
		builder.setBolt("DataStore", new DataStore(), 1)
				.shuffleGrouping("PacketBodyParse","dataStream");
        Config conf = new Config();
        //conf.setDebug(true);
        
        if(args!=null && args.length > 0) {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("test", conf, builder.createTopology());
            Utils.sleep(900000);
            cluster.killTopology("test");
            cluster.shutdown();
        }
	}

}
